package com.zmzncs.lmtc.common.config;

import com.zmzncs.lmtc.common.util.CallbackUtil;
import com.zmzncs.lmtc.common.util.PropertiesUtil;
import com.zmzncs.lmtc.module.park.dao.ParkInfoDao;
import com.zmzncs.lmtc.module.park.pojo.parkinfo.ParkInfoDTO;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import javax.validation.Valid;
import java.util.List;
import java.util.stream.Collectors;

@Slf4j
@Component
public class ClientConfig implements ApplicationRunner {

    @Autowired
    private CallbackUtil callbackUtil;
    @Autowired
    private ParkInfoDao parkInfoDao;

    @Value("${spring.mqtt.clientId}")
    private String clientId;

    private static MqttClient client;
    public static MqttConnectOptions options = new MqttConnectOptions();


    @Override
    public void run(ApplicationArguments args) {
        log.info("初始化并启动......");
        this.connect();
    }

    /**
     *  创建客户端  --- 1 ---
     */
    public MqttClient getClient() {
        try {
            clientId = clientId + "_" + (int)(Math.random()*10000);
            if (null == client) {
                client = new MqttClient(PropertiesUtil.MQTT_HOST, clientId, new MemoryPersistence());
            }
            log.info("--创建客户端【{}】", clientId);
        } catch (Exception e) {
            log.error("创建客户端异常：" + e);
        }

        return client;
    }

    /**
     * 连接mqtt服务器
     */
    private void connect() {
        try {
            // 1 创建客户端
            getClient();
            // 2 设置配置
            log.info("--生成配置对象");
            options.setUserName(PropertiesUtil.MQTT_USER_NAME);
            options.setPassword(PropertiesUtil.MQTT_PASSWORD.toCharArray());
            // 设置超时时间
            options.setConnectionTimeout(PropertiesUtil.MQTT_TIMEOUT);
            // 设置会话心跳时间
            options.setKeepAliveInterval(PropertiesUtil.MQTT_KEEP_ALIVE);
            // 是否清除session
            options.setCleanSession(false);

//            String[] topics = PropertiesUtil.MQTT_TOPIC.split(",");
            List<ParkInfoDTO> parkInfoDTOList = parkInfoDao.getByParkSource(4);
            List<String> bolianParkIdList = parkInfoDTOList.stream().map(ParkInfoDTO::getBolianParkId).collect(Collectors.toList());
            String[] topics = new String[bolianParkIdList.size()];
            for(int i=0; i<bolianParkIdList.size(); i++){
                log.info("加载停车场【{}】【{}】", parkInfoDTOList.get(i).getParkName(), bolianParkIdList.get(i));
                topics[i] = "tpc/" + bolianParkIdList.get(i);
            }

            // 3 消息发布质量
            int[] qos = getQos(topics.length);
            // 4 最后设置
            log.info("添加回调处理类");
            client.setCallback(callbackUtil);
            client.connect(options);

            //订阅消息
            log.info("订阅主题：【{}】", topics);
            client.subscribe(topics, qos);
        } catch (Exception e) {
            log.error("连接异常：" + e);
        }
    }

    /**
     *  qos   --- 3 ---
     */
    public int[] getQos(int length) {

        int[] qos = new int[length];
        for (int i = 0; i < length; i++) {
            /**
             *  MQTT协议中有三种消息发布服务质量:
             *
             * QOS0： “至多一次”，消息发布完全依赖底层 TCP/IP 网络。会发生消息丢失或重复。这一级别可用于如下情况，环境传感器数据，丢失一次读记录无所谓，因为不久后还会有第二次发送。
             * QOS1： “至少一次”，确保消息到达，但消息重复可能会发生。
             * QOS2： “只有一次”，确保消息到达一次。这一级别可用于如下情况，在计费系统中，消息重复或丢失会导致不正确的结果，资源开销大
             */
            qos[i] = 1;
        }
        log.info("设置消息发布质量");
        return qos;
    }

    /**
     * 发布，非持久化
     *
     *  qos根据文档设置为1
     */
    public void publish(String topic, String serviceId, String message) {
        MqttMessage mqttMessage = new MqttMessage();
        mqttMessage.setQos(1);
        mqttMessage.setRetained(false);
        mqttMessage.setPayload(message.getBytes());
        MqttTopic mqttTopic = client.getTopic(topic);

        if (null == mqttTopic) {
            log.error("主题：" + topic + " 不存在");
        } else {
            try {
                MqttDeliveryToken token = mqttTopic.publish(mqttMessage);
                token.waitForCompletion();

                //  不打印云平台心跳
//                if(!topic.equals("cloud/internet/time")){
                    log.info("发送消息【{}】【{}】", serviceId, mqttMessage);
//                }
            } catch (MqttException e) {
                e.printStackTrace();
            }
        }

    }

}
